package com.atguigu.realtime.dws

import com.atguigu.realtime.BaseAppV3
import com.atguigu.realtime.bean.{OrderDetail, OrderInfo, OrderWide}
import com.atguigu.realtime.util.MyRedisUtil
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.OffsetRange
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.json4s.jackson.JsonMethods
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

/**
 * Author atguigu
 * Date 2020/11/20 9:09
 */
object DwsOrderWideApp extends BaseAppV3 {
    override val master: String = "local[2]"
    override val appName: String = "DwsOrderWideApp"
    override val groupId: String = "DwsOrderWideApp"
    override val topics: Seq[String] = Seq("dwd_order_info", "dwd_order_detail")
    override val bachTime: Int = 5
    
    override def run(ssc: StreamingContext,
                     sourceStreams: Map[String, DStream[ConsumerRecord[String, String]]],
                     offsetRanges: Map[String, ListBuffer[OffsetRange]]): Unit = {
        val orderInfoStream = sourceStreams("dwd_order_info")
            .map(record => {
                implicit val f = org.json4s.DefaultFormats + toLong + toDouble
                val orderInfo: OrderInfo = JsonMethods.parse(record.value()).extract[OrderInfo]
                (orderInfo.id, orderInfo)
            })
            .window(Seconds(bachTime * 5), Seconds(bachTime))
        val orderDetailStream = sourceStreams("dwd_order_detail")
            .map(record => {
                implicit val f = org.json4s.DefaultFormats + toLong + toDouble
                val orderDetail: OrderDetail = JsonMethods.parse(record.value()).extract[OrderDetail]
                (orderDetail.order_id, orderDetail)
            })
            .window(Seconds(bachTime * 5), Seconds(bachTime))
        
        
        // 1. 对两个流进行join
        val orderWideStream: DStream[OrderWide] = orderInfoStream
            .join(orderDetailStream)
            .map {
                case (orderId, (orderInfo, orderDetail)) =>
                    //OrderWide().mergeOrderInfo(orderInfo).mergeOrderDetail(orderDetail)
                    new OrderWide(orderInfo, orderDetail)
            }
        
        // 2. 去重: 借助于 redis来去重. 读写redis:  用Set(order_detail_id)
        val orderWideDStreamDistinct: DStream[OrderWide] = orderWideStream.mapPartitions(orderWideIt => {
            val client: Jedis = MyRedisUtil.getClient
            
            val result = orderWideIt.filter(orderWide => {
                val key = s"order_join_${orderWide.order_id}"
                val r = 1 == client.sadd(key, orderWide.order_detail_id.toString)
                if (r) client.expire(key, 60 * 30)
                r
            })
            client.close()
            result
        })
        orderWideDStreamDistinct.print(1000)
        
        
        
        /*.foreachRDD(rdd => {
            println("foreachRDD...")
            rdd.collect().foreach(println)
            
            OffsetManager.saveOffsets(offsetRanges.values.reduce(_ ++ _), groupId, topics)
        })
    */
    }
}

/*
val value: DStream[(Long, (OrderInfo, OrderDetail))] = orderInfoStream.join(orderDetailStream)
val value1: DStream[(Long, (OrderInfo, Option[OrderDetail]))] = orderInfoStream.leftOuterJoin(orderDetailStream)
val value2: DStream[(Long, (Option[OrderInfo], OrderDetail))] = orderInfoStream.rightOuterJoin(orderDetailStream)
val value3: DStream[(Long, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoStream.fullOuterJoin(orderDetailStream)


// 一个key存在, 并且没有设置过过期时间   保证一个key值设置一次过期时间
                if(client.exists(key) && client.ttl(key) == -1){
                    client.expire(key, 60 * 30)
                }
 */